package com.alpaca.recommend.mapreduce;

import com.alpaca.recommend.bean.TagItem;
import com.alpaca.recommend.bean.UItem;
import com.alpaca.recommend.bean.UTags;
import com.alpaca.recommend.hbase.HTableClients;
import com.alpaca.recommend.util.MD5Util;
import com.alpaca.recommend.util.SpringContextLoader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.concurrent.Future;

/**
 * Created by justin on 15-2-10.
 */
@Component
public class CalculateUItemsRank {


    static class CalculateUItemMapper extends TableMapper<Text,IntWritable> {

        private Table tagItemsTable;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            tagItemsTable = SpringContextLoader.getBean(HTableClients.class).getHTable(TagItem.TABLE_NAME);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
//            tagItemsTable.flushCommits();
        }

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            String uTagsRowKey = Bytes.toString(key.get());
            String channelcodeHash = uTagsRowKey.substring(0,16);
            String userIdHash = uTagsRowKey.substring(16,32);
            String tag = uTagsRowKey.substring(32);
            int uTagRank = Bytes.toInt(CellUtil.cloneValue(value.getColumnLatestCell(Bytes.toBytes(UTags.U_TAGS_FAMILY), Bytes.toBytes(UTags.U_TAGS_T_RANK))));
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(channelcodeHash + MD5Util.md5(tag)));
            scan.setStopRow(Bytes.toBytes(channelcodeHash + MD5Util.md5(tag) + 'g'));
            ResultScanner rs = tagItemsTable.getScanner(scan);
            for(Result result:rs){
                Cell cell = result.getColumnLatestCell(TagItem.TAG_ITEM_FAMILY,TagItem.TAG_ITEM_RANK_COLUMN);
                int rank = Bytes.toInt(CellUtil.cloneValue(cell));
                String itemId = Bytes.toString(CellUtil.cloneRow(cell)).substring(32);
                String rowKey = userIdHash+channelcodeHash+itemId;
                context.write(new Text(rowKey) ,new IntWritable(rank*uTagRank));
            }
        }
    }

    static class CalculateUItemReducer extends TableReducer<Text,IntWritable , ImmutableBytesWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int totalRank = 0;
            for(IntWritable rank : values){
                totalRank += rank.get();
            }
            new UItem().setRow(key.toString()).set(
                    UItem.U_ITEM_FAMILY,UItem.U_ITEM_RANK_COLUMN,totalRank
            ).save();
        }
    }

    public static Boolean excuteJob()throws IOException, ClassNotFoundException, InterruptedException{
        Configuration conf = new HTableClients().getConf();
        String table = "u_tags";
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(UTags.U_TAGS_FAMILY),Bytes.toBytes(UTags.U_TAGS_T_RANK));
        Job job = Job.getInstance(conf,"Calculate to u_items rank");
        job.setJarByClass(CalculateUItemsRank.class);
        TableMapReduceUtil.initTableMapperJob(table,scan,CalculateUItemMapper.class,Text.class,IntWritable.class,job);
        job.setOutputFormatClass(NullOutputFormat.class);
        job.setNumReduceTasks(0);
        return job.waitForCompletion(true)?true:false;
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        SpringContextLoader.init();
        Configuration conf = new HTableClients().getConf();
        String table = "u_tags";
        String outputTable = "u_items";
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(UTags.U_TAGS_FAMILY),Bytes.toBytes(UTags.U_TAGS_T_RANK));
        Job job = Job.getInstance(conf,"Calculate to u_items rank");
        job.setJarByClass(CalculateUItemsRank.class);
        TableMapReduceUtil.initTableMapperJob(table,scan,CalculateUItemMapper.class,Text.class,IntWritable.class,job);
        TableMapReduceUtil.initTableReducerJob(
                outputTable,        // output table
                CalculateUItemReducer.class,    // reducer class
                job);
        job.setNumReduceTasks(1);
        System.exit(job.waitForCompletion(true)?0:1);
    }
}
